package org.example.flink.connections.pulsar;

import cn.hutool.core.date.DateUtil;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.pulsar.FlinkPulsarSource;
import org.apache.flink.streaming.util.serialization.PulsarDeserializationSchema;

import java.util.Properties;

public class TestFlinkSource {

    public static void main(String[] args) throws Exception{


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);
        env.setParallelism(1);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        String serviceUrl = "pulsar://127.0.0.1:6650";
        String adminUrl = "http://127.0.0.1:8080";

        Properties props=new Properties();
        props.setProperty("topic","persistent://public/default/delay10");
        props.setProperty("partition.discovery.interval-millis", "5000");
        props.setProperty("pulsar.reader.subscriptionName", "s1");
        props.setProperty("subscriptionType", "Shared");
        props.setProperty("consumerName", "c1");
        props.setProperty("subscriptionMode", "Durable");

        FlinkPulsarSource<String> source = new FlinkPulsarSource<>(serviceUrl, adminUrl,
                PulsarDeserializationSchema.valueOnly(new SimpleStringSchema()), props);

        // or setStartFromLatest、setStartFromSpecificOffsets、setStartFromSubscription
        source.setStartFromEarliest();

        DataStream<String> stream = env.addSource(source);

        stream.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                System.out.println("============="+value+" now="+ DateUtil.now());
            }
        });

        env.execute("TestFlinkSource");

    }

}
